﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.Data.Sql;
using System.Data.SqlClient;
using System.Threading.Tasks;
using System.Security.Cryptography;
using System.Transactions;
using Crypto;
using SQLDBManager;

namespace SQLCDCIDLoader
{
    class SQLCDCDAL
    {

        public string SourceConnectionString { get; set; }
        public string DestConnectionString { get; set; }
        public string SQLJobID { get; set; }
        public string srcserver { get; set; }
        public string userid { get; set; }
        public string epwd { get; set; }        
        
        
        public int cnid;
        private string srctable="";
        private string desttable="";
        private string srcdb = "";
        private string destdb = "";
        private string _destserver = "";



        LogRoller lr = new LogRoller();
        
        
        public SQLCDCDAL(string _sqljobid,string _srcserver,string _userid,string _epwd,string _logfile)
        {
            //LogRoller lgr = new LogRoller(_logfile);
            lr.logpath = _logfile;    
            SQLJobID = _sqljobid;
            srcserver = _srcserver;
            userid = _userid;
            try
            {
                if (!string.IsNullOrEmpty(_epwd) || _epwd == "")
                { epwd = Crypto.Crypto.Decrypt(_epwd, true); }
                fn_getconnectionstring();
                
            }catch(Exception)
            {
                
                throw;
            }
        }

        /// <summary>
        /// initialise source and destination connection string.
        /// </summary>
        /// <returns></returns>
        private void fn_getconnectionstring()
        {
            try
            {
                string _sqlqry = "select cnid,srcserver,destserver,srcdb,destdb,srctable,desttable,userid,password from "+
                " SQLCDCApp.sqlcdcapp.connection_details where " +
                " SQLJobID='" + SQLJobID + "' and Enabled=1";

                string _constr = "";
                if (userid == "$$@#$$")
                {
                    _constr = "Initial Catalog=master;Data Source=" + srcserver + ";Integrated Security=True;";
                }
                else
                {
                    _constr = "Initial Catalog=master;Data Source=" + srcserver + ";user id=" + userid + ";password=" + epwd;
                }

                QueryExecution qe = new QueryExecution(_constr);
                //DataTable dtresults = fn_ExecuteReader(_sqlqry, _constr);
                DataTable dtresults = qe.fn_ExecuteTable(_sqlqry);

                if (dtresults.Rows.Count == 0)
                {
                    // log error msg and quit
                    // 
                }

                srcdb = dtresults.AsEnumerable().Select(s => s.Field<string>("srcdb")).FirstOrDefault<string>();
                srctable = dtresults.AsEnumerable().Select(s => s.Field<string>("srctable")).FirstOrDefault<string>();
                cnid = dtresults.AsEnumerable().Select(s => s.Field<int>("cnid")).FirstOrDefault<int>();
                _destserver = dtresults.AsEnumerable().Select(s => s.Field<string>("destserver")).FirstOrDefault<string>();
                destdb = dtresults.AsEnumerable().Select(s => s.Field<string>("destdb")).FirstOrDefault<string>();
                desttable = dtresults.AsEnumerable().Select(s => s.Field<string>("desttable")).FirstOrDefault<string>();
                string _destuserid = dtresults.AsEnumerable().Select(s => s.Field<string>("userid")).FirstOrDefault<string>();
                string _destpwd = dtresults.AsEnumerable().Select(s => s.Field<string>("password")).FirstOrDefault<string>();
                if (!string.IsNullOrEmpty(_destpwd))
                {
                    _destpwd = Crypto.Crypto.Decrypt(_destpwd, true);
                }

                if (userid == "$$@#$$")
                {
                    SourceConnectionString = "Initial Catalog=" + srcdb + ";Data Source=" + srcserver + ";Integrated Security=True;";
                }
                else
                {
                    SourceConnectionString = "Initial Catalog=" + srcdb + ";Data Source=" + srcserver + ";user id=" + userid + ";password=" + epwd;

                }

                if (userid == "$$@#$$")
                {
                    DestConnectionString = "Initial Catalog=" + destdb + ";Data Source=" + _destserver + ";Integrated Security=True;";
                }
                else
                {
                    DestConnectionString = "Initial Catalog=" + destdb + ";Data Source=" + _destserver + ";user id=" 
                        + _destuserid + ";password=" + _destpwd;

                }

            }catch(Exception)
            {
                throw;
            }
        }
        
        /// <summary>
        /// perform incremental load
        /// </summary>
        public void fn_IncrementalDataLoad()
        {
            try
            {
                
                lr.LogMessage(srcserver + "." + srcdb + "." + srctable + " => " + _destserver + "." + destdb + "." + desttable);

                //fn_ExecuteQuery("insert into logpath values('" + lr.logpath.ToString() + "')", SourceConnectionString);
                string _pkquery, _delqry = "";
                //string _pkdatatype, _updatesql = "";
                string _QryCurrentMaxLsn, _createtmptable, _qry, _usercolqry;
                //string _sqlqrycols = " SET ";
                DataTable dtinserts = new DataTable();
                DataTable dtdeletes = new DataTable();
                DataTable dtupdates = new DataTable();
                QueryExecution sqe = new QueryExecution(SourceConnectionString);
                QueryExecution dqe = new QueryExecution(DestConnectionString);
                #region Getchangedata
                _usercolqry = "select stuff((select ',cte.'+ name from sys.columns where object_id=object_id('" + srctable + "') " +
                   "order by column_id for xml path('')),1,1,'') ";

                //string _usercols = fn_ExecuteScalor(_usercolqry, SourceConnectionString);
                string _usercols = sqe.fn_ExecuteScalor(_usercolqry);

                string _captureinstanceqry = "Select captureinstance,CONVERT(BINARY(10), srclaststartlsn, 1) from " + 
                    " sqlcdcapp.sqlcdcapp.IDLoad where SQLJobID='" + SQLJobID + "' order by srclastsyncdate desc";

                //DataTable _dtresults = fn_ExecuteReader(_captureinstanceqry, SourceConnectionString);
                DataTable _dtresults = sqe.fn_ExecuteTable(_captureinstanceqry);
                string _captureinstance = _dtresults.AsEnumerable().Select(s => s.Field<string>("captureinstance")).FirstOrDefault<string>();

                _QryCurrentMaxLsn = "Select UPPER(sys.fn_varbintohexstr(MAX(__$start_lsn))) FROM cdc." + srctable.Replace(".", "_") + "_CT";

                string _CurrentMaxLsn = sqe.fn_ExecuteScalor(_QryCurrentMaxLsn);
                // get change data into datatable

                _qry = " DECLARE @lastMaxLsn BINARY(10), @newmaxlsn  BINARY(10) " +
                  "SELECT @newmaxlsn = MAX(__$start_lsn) from cdc." + srctable.Replace(".", "_") + "_CT " +
                 " SELECT	top 1 @lastMaxLsn=CONVERT(BINARY(10), srclaststartlsn, 1) FROM   sqlcdcapp.sqlcdcapp.idload " +
                 " WHERE  captureinstance = '" + _captureinstance + "' ORDER  BY srclastsyncdate DESC " +
                 "if(@lastMaxLsn is NULL and @newmaxlsn is not NULL)" +
                 "  begin set @lastMaxLsn = sys.fn_cdc_get_min_lsn('" + _captureinstance + "') end " +
                 " select  " + _usercols + " , CASE cte.__$Operation When 1 THEN 'DELETE' WHEN  2 THEN 'INSERT' WHEN 3 THEN 'UPDATE BEFORE'  " +
                 " WHEN 4 THEN 'UPDATE AFTER'  END AS Operation ," +
                 " Stuff((SELECT    ','+CC.column_name FROM      cdc.captured_columns CC INNER JOIN " +
                 " cdc.change_tables CT ON CC.[object_id] = CT.[object_id] WHERE     capture_instance = '" + _captureinstance + "' " +
                 " AND sys.fn_cdc_is_bit_set(CC.column_ordinal, cte.__$update_mask) = 1 FOR XML PATH('')),1,1,'') As ColumnsUpdated " +
                 " from cdc.fn_cdc_get_net_changes_" + srctable.Replace(".", "_") + "(@lastMaxLsn,@newmaxlsn ,'all with mask') cte " +
                    // " join  cdc.dbo_employee_CT xyz on cte.__$start_lsn=xyz.__$start_lsn " +
                 " where cte.__$start_lsn> @lastMaxLsn and cte.__$start_lsn < = @newmaxlsn";

              //  DataTable dtchangedata = fn_ExecuteReader(_qry, SourceConnectionString);
                DataTable dtchangedata = sqe.fn_ExecuteTable(_qry);

                _pkquery = " SELECT column_name FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE " +
                           " WHERE OBJECTPROPERTY(OBJECT_ID(constraint_schema + '.' + constraint_name), 'IsPrimaryKey') = 1 AND " +
                           " table_name = '" + srctable.Substring(srctable.IndexOf(".") + 1) + "'" +
                           " and TABLE_SCHEMA ='" + srctable.Substring(0, srctable.IndexOf(".")) + "'";
                
               // string _pkcolumn = fn_ExecuteScalor(_pkquery, SourceConnectionString);
                DataTable _dtpkcols = sqe.fn_ExecuteTable(_pkquery);
               // string _pkcolumn = sqe.fn_ExecuteScalor(_pkquery);
              //  _pkdatatype = dtchangedata.Columns[_pkcolumn].DataType.ToString();
                
                string _pkcollist = string.Join(",", _dtpkcols.Rows.OfType<DataRow>().Select(r => r[0]));
                #endregion

                using (TransactionScope tran = new TransactionScope())
                {
                    #region ApplyInserts
                    var inserts = from myRow in dtchangedata.AsEnumerable()
                                  where myRow.Field<string>("Operation") == "INSERT"
                                  select myRow;
                    if (inserts.Count() > 0)
                    {
                        dtinserts = inserts.AsDataView().ToTable();
                        dtinserts.Columns.Remove("ColumnsUpdated");
                        // do bulk insert into destination table

                        using (SqlConnection _sqlcon = new SqlConnection(DestConnectionString))
                        {
                            _sqlcon.Open();
                            using (SqlBulkCopy sbc = new SqlBulkCopy(_sqlcon))
                            {
                                sbc.DestinationTableName = desttable;
                                foreach (var column in dtinserts.Columns)
                                    sbc.ColumnMappings.Add(column.ToString(), column.ToString());
                                sbc.WriteToServer(dtinserts);
                            }
                        }
                        
                    }
                    lr.LogMessage(dtinserts.Rows.Count.ToString() + " rows inserted in table " + desttable);
                    #endregion

                    #region ApplyDeletes

                    var deletes = from myRow in dtchangedata.AsEnumerable()
                                  where myRow.Field<string>("Operation") == "DELETE"
                                  select myRow;

                    if (deletes.Count() > 0)
                    {
                        dtdeletes = deletes.AsDataView().ToTable(false, _pkcollist.Split(','));
                        //string _pkcols = string.Join(", ", Array.ConvertAll(_dtpkcols.Select(), i => i.ToString())); 
                        

                        //string dtable = "sqlcdctmp_" + desttable.Substring(desttable.IndexOf(".") + 1);
                        //_createtmptable = "IF OBJECT_ID('" + dtable + "'" + ") IS NOT NULL BEGIN DROP TABLE " + dtable + " END " +
                        //" Select " + _pkcolumn + " into " + dtable + " FROM " + desttable + " where 1=2 ";

                        string dtable = "sqlcdctmp_" + desttable.Substring(desttable.IndexOf(".") + 1);
                        _createtmptable = "IF OBJECT_ID('" + dtable + "'" + ") IS NOT NULL BEGIN DROP TABLE " + dtable + " END " +
                        " Select " + _pkcollist + " into " + dtable + " FROM " + desttable + " where 1=2 ";
                        
                        
                        dqe.fn_ExecuteQuery(_createtmptable);
                        //fn_ExecuteQuery(_createtmptable, DestConnectionString);

                        // dump primary key column in the temp table
                        using (SqlConnection _sqldestcon = new SqlConnection(DestConnectionString))
                        {
                            _sqldestcon.Open();
                            using (SqlBulkCopy sbc = new SqlBulkCopy(_sqldestcon))
                            {
                                sbc.DestinationTableName = dtable;
                                foreach (var column in dtdeletes.Columns)
                                    sbc.ColumnMappings.Add(column.ToString(), column.ToString());
                                sbc.WriteToServer(dtdeletes);
                            }
                        }
                        string _joincols = "";
                        // create delete query - composite primary key
                        foreach(DataRow _dr in _dtpkcols.Rows)
                        {
                            //_matchpkcols = _matchpkcols + _dr[0].ToString();
                            _joincols = _joincols + "dt." + _dr[0].ToString() + " = " + "tmpdt." + _dr[0] + " and ";

                        }
                        //_delqry = " Delete dt from " + desttable + " dt join " + dtable + " tmpdt ON " +
                          //  " dt." + _pkcolumn + " = " + " tmpdt." + _pkcolumn;

                        _delqry = " Delete dt from " + desttable + " dt join " + dtable + " tmpdt ON " +
                            _joincols.Substring(0, _joincols.LastIndexOf("and"));
                        

                        dqe.fn_ExecuteQuery(_delqry);

                        // delete the temp table
                        dqe.fn_ExecuteQuery("IF OBJECT_ID('" + dtable + "'" + ") IS NOT NULL BEGIN DROP TABLE " + dtable + " END ");
                    }
                    lr.LogMessage(dtdeletes.Rows.Count.ToString() + " rows deleted from table " + desttable);
                    #endregion

                    #region Apply Updates
                    // dump update data in temp table at destination
                    // join temp table to destination table to update the records.
                    // get updated records
                    var
                        _updates = from myRow in dtchangedata.AsEnumerable()
                                   where myRow.Field<string>("Operation") == "UPDATE AFTER"
                                   select myRow;


                    DataTable dtupdate = _updates.AsDataView().ToTable();
                    //string[] _columnstoupdate;

                    if (_updates.Count() > 0)
                    {
                        
                        // get primary key cols
                        string dtable = "sqlcdctmp_" + desttable.Substring(desttable.IndexOf(".") + 1);
                        _createtmptable = "IF OBJECT_ID('" + dtable + "'" + ") IS NOT NULL BEGIN DROP TABLE " + dtable + " END " +
                        " Select " + _pkcollist + " into " + dtable + " FROM " + desttable + " where 1=2 ";
                        string _columnstoupdate = "";
                        // get list of columns updated 
                        foreach (DataRow _dr in dtupdate.Rows)
                        {
                            _columnstoupdate = _dr["ColumnsUpdated"].ToString();
                        }

                        _createtmptable = "IF OBJECT_ID('" + dtable + "'" + ") IS NOT NULL BEGIN DROP TABLE " + dtable + " END " +
                       " Select " + _pkcollist + "," + _columnstoupdate + " into " + dtable + " FROM " + desttable + " where 1=2 ";
                        dqe.fn_ExecuteQuery(_createtmptable);
                        dtupdates = _updates.AsDataView().ToTable(false, (_pkcollist+ "," + _columnstoupdate).Split(','));
                        // dump primary key column in the temp table
                        using (SqlConnection _sqldestcon = new SqlConnection(DestConnectionString))
                        {
                            _sqldestcon.Open();
                            using (SqlBulkCopy sbc = new SqlBulkCopy(_sqldestcon))
                            {
                                sbc.DestinationTableName = dtable;
                                foreach (var column in dtupdates.Columns)
                                    sbc.ColumnMappings.Add(column.ToString(), column.ToString());
                                sbc.WriteToServer(dtupdates);
                            }
                        }

                        // prepare update query
                        // update dt  SET dt.col2=tmp.col2  FROM tblcomkey dt join sqlcdctmp_cdc_tblcomkey tmp 
                        // ON dt.col1=tmp.col1 and dt.sno=tmp.sno
                        string[] _updatecollist = _columnstoupdate.Split(',');
                        string _qrySetColumns = "";
                        string _updateqry = " UPDATE dt SET ";
                        
                        // prepare update-set columns
                        foreach(string _item in _updatecollist)
                        {
                            _qrySetColumns = _qrySetColumns + " dt." + _item + " = " + " tmpdt." + _item + " , ";
                                
                        }
                        string _joincols = "";
                        // prepare join columns 

                        foreach (DataRow _dr in _dtpkcols.Rows)
                        {
                            //_matchpkcols = _matchpkcols + _dr[0].ToString();
                            _joincols = _joincols + "dt." + _dr[0].ToString() + " = " + "tmpdt." + _dr[0] + " and ";

                        }
                        
                        // the update query
                        _updateqry = _updateqry + _qrySetColumns.Substring(0, _qrySetColumns.LastIndexOf(','))
                            + " FROM " + desttable + " dt JOIN " + dtable + " tmpdt ON "
                            + _joincols.Substring(0, _joincols.LastIndexOf("and"));


                        dqe.fn_ExecuteQuery(_updateqry);

                        // delete the temp table
                        dqe.fn_ExecuteQuery("IF OBJECT_ID('" + dtable + "'" + ") IS NOT NULL BEGIN DROP TABLE " + dtable + " END ");
                       
                    }

                    lr.LogMessage(dtupdate.Rows.Count.ToString() + " rows updated in table " + desttable);
                    #endregion


                    #region UpdateMetaTable
                    if(string.IsNullOrEmpty(_CurrentMaxLsn))
                    {
                        _CurrentMaxLsn = DBNull.Value.ToString();
                    }
                        string _sqlqry = "INSERT INTO sqlcdcapp.sqlcdcapp.IDLoad VALUES('" + _captureinstance + "','" + _CurrentMaxLsn + "'," +
                      " Getdate(),suser_name(),'" + SQLJobID + "'," + dtinserts.Rows.Count + "," + dtdeletes.Rows.Count 
                      + "," + dtupdate.Rows.Count + ")";
                    //update idload tables with load details
                    sqe.fn_ExecuteQuery(_sqlqry);
                    lr.LogMessage("Meta tables updated.");
                    #endregion

                    tran.Complete();


                }
                lr.LogMessage("Incremental load completed " + srcserver + "." + srcdb + "." + srctable + " => " + _destserver + "." + 
                    destdb + "." + desttable);
            }catch(Exception ex)
            {
                lr.LogException(ex);
            }
            
        }



    }


    
}
